package task

import (
	"cm2/exception"
	"cm2/exception/taskexception"
	"cm2/service/task/jobs"
	"cm2/service/task/tasks"
	"cm2/service/task/util"
	"encoding/json"
	"fmt"
	"os"
	"runtime"
	"time"

	"github.com/go-xorm/xorm"

	"github.com/kataras/golog"
)

// JobService Job Service
type JobService struct {
	SId         string
	Name        string
	Jobs        map[string]jobs.Jobprovider
	Tasks       map[string]tasks.Taskprovider
	jbp         *xorm.Engine
	concurrency int
	cLock       chan bool
	queen       int
	logger      *golog.Logger
}

// NewJobService 初始化 Job Serivce
func NewJobService(sid string, name string) *JobService {
	js := &JobService{}
	js.Jobs = jobs.DefaultJobs()
	js.Tasks = tasks.DefaultTasks()
	js.Name = name
	js.SId = sid

	return js
}

// SetJobLogProvider 设置Job日志的处理程序
func (js *JobService) SetJobLogProvider(jbp *xorm.Engine) {
	js.jbp = jbp
}

// SetLogger 设置日志
func (js *JobService) SetLogger(logger *golog.Logger) {
	js.logger = logger
}

// Logger 返回日志记录器
func (js *JobService) Logger() *golog.Logger {
	if js.logger == nil {
		js.logger = golog.New()
		js.logger.SetOutput(os.Stdout)
	}

	return js.logger
}

// SetConcurrency 设置并行数量，客户端指 Job 的并行数量
func (js *JobService) SetConcurrency(c int) {
	if c <= 0 {
		c = 1
	}
	js.Logger().Infof("job service concurrency set to %d", c)
	js.concurrency = c
}

// Concurrency 返回并行数量设置
func (js *JobService) Concurrency() int {
	return js.concurrency
}

// SetMaxProc 设置最大使用的CPU内核数，在服务器性能足够，且jobservice性能不足的情况下增加
func (js *JobService) SetMaxProc(n int) {
	cn := runtime.NumCPU()
	if n > cn {
		n = cn
	}
	runtime.GOMAXPROCS(n)
	js.Logger().Infof("job service set runtime max process to %d", n)
}

// ResolveJob 通过对应的Job处理程序解析任务，并获取完整的task清单
func (js *JobService) ResolveJob(jobSId string, params map[string]interface{}) (*jobs.Job, *exception.Exception) {
	jbp, ok := js.Jobs[jobSId]
	if !ok {
		return nil, exception.NewExceptionWithDefine(taskexception.JobProviderNotFound, jobSId)
	}

	tasks, err := jbp.ResolveJob(params)
	if err != nil {
		return nil, exception.NewExceptionWithDefine(taskexception.GenerateTasksFailed, err)
	}

	ji := &jobs.Job{
		Serial: util.GenSerial(jobSId),
		SId:    jobSId,
		Code:   jbp.InnerJob().Code,
		Name:   jbp.InnerJob().Name,
		Tasks:  tasks,
	}

	return ji, nil
}

// ResolveReport 解析报告（服务端）
func (js *JobService) ResolveReport(rpt *Report, err error) {
	if err != nil {
		// TODO： 这里好像暂时无法回报错误，因为无法对应到具体的任务，需要调整通信模型
		js.Logger().Errorf("unable to resolve report: %v", err)
		return
	}
	// 暂时只处理些日志
	mode := rpt.Mode
	data := rpt.Data.(map[string]interface{})
	ts := int64(data["UpdatedOn"].(float64))

	js.UpdateJobTaskLogState(
		mode,
		data["Serial"].(string),
		int(data["State"].(float64)),
		data["Message"].(string),
		time.Unix(ts, 0),
	)
}

// BeginExecuteJob 开始执行Job（客户端）
func (js *JobService) BeginExecuteJob(job *jobs.JobConv, rptFn func(*Report) *exception.Exception, err error) {
	if err != nil {
		// TODO： 这里好像暂时无法回报错误，因为无法对应到具体的任务，需要调整通信模型
		js.Logger().Errorf("unable to resolve job: %v", err)
		return
	}
	now := time.Now()
	go func() {
		// 3、C 客户端接收 Job 全部记录写入 Job 和 Task
		js.CreateJobLog(job.Serial, job.SId, job.Code, job.Name, job.Tasks, JobAccepted)
		now = time.Now()
		// 4、S 客户端回报服务端已接收 Job
		js.UpdateLogSyncState(JobExecuteReport, job.Serial, rptFn(NewReport(JobExecuteReport, job.Serial, JobAccepted, "", now)))

		now = time.Now()
		// 5、C 客户端检查 Job 并发量
		js.UpdateJobTaskLogState(JobExecuteReport, job.Serial, JobPending, "", now)
		// 6、S 客户端回报
		js.UpdateLogSyncState(JobExecuteReport, job.Serial, rptFn(NewReport(JobExecuteReport, job.Serial, JobPending, "", now)))
		// 	Job 待执行 3，Task 待执行 0

		queened := false
		if len(js.cLock) >= js.concurrency {
			js.Logger().Infof("reach max concurrency %d, job %s %s push into queen", js.concurrency, job.Code, job.Serial)
			js.queen++
			queened = true
		}
		// 尝试写入一个信道量，如果写入不了就阻塞了...
		js.cLock <- true
		if queened {
			js.Logger().Infof("job %s %s restore", job.Code, job.Serial)
			js.queen--
		}
		js.Logger().Infof("current concurrency %d, %d in queen", len(js.cLock), js.queen)

		// 执行task
		msg, err := js.executeJob(job, rptFn)
		now = time.Now()
		// task 轮训完，回报Job
		if err != nil {
			js.Logger().Error("job %s %s failed: %s %v", job.Code, job.Serial, msg, err)
			//  13.2 失败 Job 失败 99 异常描述
			em := fmt.Sprintf("%s\n%v", msg, err)
			js.UpdateJobTaskLogState(JobExecuteReport, job.Serial, JobError, em, now)
			// 回报 14.2 失败 Job 失败 99 异常描述
			js.UpdateLogSyncState(JobExecuteReport, job.Serial, rptFn(NewReport(JobExecuteReport, job.Serial, JobError, em, now)))

		} else {
			js.Logger().Infof("job %s %s complete", job.Code, job.Serial)
			// 13.1 成功 Job 成功 5
			js.UpdateJobTaskLogState(JobExecuteReport, job.Serial, JobComplete, "", now)
			// 回报 14.1 成功 Job 成功 5
			ex := rptFn(NewReport(JobExecuteReport, job.Serial, JobComplete, "", now))
			if ex == nil {
				// 回报成功后
				js.RemoveJobLog(job.Serial)
			}

		}
		// 退并发信道
		<-js.cLock
		js.Logger().Infof("current concurrency %d, %d in queen", len(js.cLock), js.queen)
	}()
}

// ExecuteJob 执行 Job
func (js *JobService) executeJob(job *jobs.JobConv, rptFn func(*Report) *exception.Exception) (string, *exception.Exception) {
	now := time.Now()
	// 7、C 客户端执行 Job 正在执行 4，Task 待执行 0
	js.UpdateJobTaskLogState(JobExecuteReport, job.Serial, JobExecuting, "", now)
	// 8、S 客户端回报
	js.UpdateLogSyncState(JobExecuteReport, job.Serial, rptFn(NewReport(JobExecuteReport, job.Serial, JobExecuting, "", now)))

	var err *exception.Exception
	var msg string
	var result interface{}
	// 遍历执行任务
	for idx, t := range job.Tasks {
		// 获取执行器
		provider, ok := js.Tasks[t.SId]
		if !ok {
			err = exception.NewExceptionWithDefine(taskexception.TaskProviderNotFound, t.SId)
			js.Logger().Errorf("execute task failed, %v", err)
			// 11.2 失败 Job 正在执行 4，Task 失败 99 异常日志 -> 任务终止
			cm := fmt.Sprintf("%v", err)
			now = time.Now()
			js.UpdateJobTaskLogState(TaskExecuteReport, t.Serial, TaskError, cm, now)
			// 12.2 失败 Job 正在执行 4，Task 失败 99 异常日志 -> 任务终止
			js.UpdateLogSyncState(TaskExecuteReport, t.Serial, rptFn(NewReport(TaskExecuteReport, t.Serial, TaskError, cm, now)))
			break
		}
		// 将上一次的结果作为下一步的输入传递
		if idx > 0 && result != nil {
			t.Params["__pretask"] = result
		}
		now = time.Now()
		// 9、C 客户端单步执行 Task Job 正在执行 4，Task 正在执行 1
		js.UpdateJobTaskLogState(TaskExecuteReport, t.Serial, TaskExecuting, "", now)
		// 10、S 客户端回报 Job 正在执行 4，Task 正在执行 1
		js.UpdateLogSyncState(TaskExecuteReport, t.Serial, rptFn(NewReport(TaskExecuteReport, t.Serial, TaskExecuting, "", now)))
		// 执行任务
		result, msg, err = provider.Execute(t)
		now = time.Now()
		if err != nil {
			js.Logger().Errorf("execute task failed, %v", err)
			// 11.2 失败 Job 正在执行 4，Task 失败 99 异常日志 -> 任务终止
			cm := fmt.Sprintf("%v", err)
			js.UpdateJobTaskLogState(TaskExecuteReport, t.Serial, TaskError, cm, now)
			// 12.2 失败 Job 正在执行 4，Task 失败 99 异常日志 -> 任务终止
			js.UpdateLogSyncState(TaskExecuteReport, t.Serial, rptFn(NewReport(TaskExecuteReport, t.Serial, TaskError, cm, now)))

			break
		} else {
			// 11.1 成功 Job 正在执行 4，Task 成功 2 ->继续轮询至任务终止
			js.UpdateJobTaskLogState(TaskExecuteReport, t.Serial, TaskComplete, msg, now)
			// 12.1 成功 Job 正在执行 4，Task 成功 2 ->继续轮询至任务终止
			js.UpdateLogSyncState(TaskExecuteReport, t.Serial, rptFn(NewReport(TaskExecuteReport, t.Serial, TaskComplete, "", now)))
		}
	}

	return msg, err
}

// CreateJobLog 创建JobLog
func (js *JobService) CreateJobLog(serial string, sid string, code string, name string, tasks []*tasks.Task, state int) {
	var err error
	// 保存Job
	job := new(JobLog)
	job.Serial = serial
	job.SId = sid
	job.Code = code
	job.Name = name
	job.State = state
	job.SyncState = SyncPending
	job.Message = ""
	if _, err = js.jbp.Insert(job); err != nil {
		js.Logger().Errorf("create job log failed: %v", err)
		return
	}
	// 循环保存 task
	if tasks != nil {
		tksArr := []*JobTaskLog{}
		for _, t := range tasks {
			var paramJSON []byte
			var e error
			if paramJSON, e = json.Marshal(t.Params); e != nil {
				paramJSON = []byte{}
			}
			tks := new(JobTaskLog)
			tks.Serial = t.Serial
			tks.SId = t.SId
			tks.Code = t.Code
			tks.Name = t.Name
			tks.Params = string(paramJSON)
			tks.State = TaskPending
			tks.SyncState = SyncPending
			tks.Message = ""
			tks.JobSerial = job.Serial
			tksArr = append(tksArr, tks)
		}
		if len(tksArr) > 0 {
			if _, err = js.jbp.Insert(tksArr); err != nil {
				js.Logger().Errorf("create job task logs failed: %v", err)
				return
			}
		}
	}
}

// RemoveJobLog 删除JobLog
func (js *JobService) RemoveJobLog(serial string) {
	var err error
	// 删除 task
	sql := "DELETE FROM `job_task_log` WHERE job_serial=?"
	if _, err = js.jbp.Exec(sql, serial); err != nil {
		js.Logger().Errorf("clean up job task logs failed: %s %v", serial, err)
	}
	// 删除 job
	sql = "DELETE FROM `job_log` WHERE serial=?"
	if _, err = js.jbp.Exec(sql, serial); err != nil {
		js.Logger().Errorf("clean up job log failed: %s %v", serial, err)
	}
}

// UpdateJobTaskLogState 更新Job和Task的状态 updatedOn 通过传递，标记为最后更新时间（即完成时间），使得服务端的数据一致
func (js *JobService) UpdateJobTaskLogState(mode string, serial string, state int, message string, updatedOn time.Time) {
	var affected int64
	var err error
	if mode == JobExecuteReport {
		job := new(JobLog)
		job.State = state
		job.Message = message
		job.UpdatedOn = updatedOn
		affected, err = js.jbp.Id(serial).Update(job)
	} else if mode == TaskExecuteReport {
		task := new(JobTaskLog)
		task.State = state
		task.Message = message
		task.UpdatedOn = updatedOn
		affected, err = js.jbp.Id(serial).Update(task)
	} else {
		js.Logger().Warnf("invalid mode for updating state: %s", mode)
		return
	}
	if err != nil || affected <= 0 {
		js.Logger().Errorf("update job task state failed: %s %s", mode, serial)
	}
}

// UpdateLogSyncState 更新Log的同步状态
func (js *JobService) UpdateLogSyncState(mode string, serial string, ex *exception.Exception) {
	var affected int64
	var err error
	syncState := SyncComplete
	if ex != nil {
		syncState = SyncError
	}
	if mode == JobExecuteReport {
		job := new(JobLog)
		job.SyncState = syncState
		job.UpdatedOn = time.Now()
		affected, err = js.jbp.Id(serial).Update(job)
	} else if mode == TaskExecuteReport {
		task := new(JobTaskLog)
		task.SyncState = syncState
		task.UpdatedOn = time.Now()
		affected, err = js.jbp.Id(serial).Update(task)
	} else {
		js.Logger().Warnf("invalid mode for updating sync state: %s", mode)
		return
	}
	if err != nil || affected <= 0 {
		js.Logger().Errorf("update job task sync state failed: %s %s %v", mode, serial, err)
	}
}

func (js *JobService) syncJobLogProvider() {
	if js.jbp == nil {
		js.Logger().Warn("job log provider is not specified")
		return
	}
	if err := js.jbp.Sync2(new(JobLog), new(JobTaskLog)); err != nil {
		js.Logger().Warnf("job log provider sync failed: %v", err)
		js.jbp = nil
	}
}

// Start 启动 TCPClient
func (js *JobService) Start() error {
	js.Logger().Infof("%s job service started, %d registered jobs and %d registered tasks found", js.Name, len(js.Jobs), len(js.Tasks))

	// 设置并发信道
	cr := js.concurrency
	if cr <= 0 {
		cr = 1
	}
	js.cLock = make(chan bool, cr)

	// 初始化数据库
	js.syncJobLogProvider()

	return nil
}

// Shutdown 发送关闭消息
func (js *JobService) Shutdown() {
	if len(js.cLock) > 0 {
		for i := 0; i < len(js.cLock); i++ {
			<-js.cLock
		}
	}
	if js.jbp != nil {
		js.jbp.Close()
	}
}
